package com.myorg.bionline.gui.control;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.myorg.domain.BlockInfo;
import com.myorg.util.DataLocalityRunner;
import com.myorg.util.DataLocalityRunner.DataLocalityTask;
import com.myorg.util.HdfsHelper;

@WebServlet("/Dispatcher")
public class Dispatcher extends HttpServlet {
	private static final long serialVersionUID = 1L;

	private static final Logger logger = LoggerFactory.getLogger(Dispatcher.class);

	public Dispatcher() {
		super();
	}

	protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
		doPost(request, response);
	}

	private static void dumpBlockInfo(List<BlockInfo> blkInfos) throws IOException {
		for (int i = 0; i < blkInfos.size(); i++) {
			String[] hosts = blkInfos.get(i).getBlockLocation().getHosts();
			String[] pathInfo = blkInfos.get(i).getBlockLocation().getTopologyPaths();
			for (int j = 0; j < pathInfo.length; j++) {
				logger.info("getTopologyPaths[" + i + "]" + pathInfo[j].toString());
			}
			logger.info("Host %d: %s %n", i, hosts);
		}
	}

	private static List<BlockInfo> generateBlockInfoList(String[] filePaths) throws IOException {
		HdfsHelper helper = new HdfsHelper();
		List<BlockInfo> rtn = new ArrayList<BlockInfo>();
		for (String path : filePaths) {
			rtn.addAll(helper.getBlockInfos(path));
		}
		return rtn;
	}

	// TODO: Move this method to Service layer
	public static void doSearch(final String key, final String[] filePaths, final PrintWriter wri,
			final String contextPath, final int port, final String syncPath) {
		// TODO: externalize the SocketTimeout parameter.
		RequestConfig requestConfig = RequestConfig.custom().setSocketTimeout(30000).setConnectTimeout(3000).build();

		final CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setDefaultRequestConfig(requestConfig)
				.build();
		try {
			List<BlockInfo> blockInfos = generateBlockInfoList(filePaths);
			dumpBlockInfo(blockInfos);

			httpclient.start();

			final CountDownLatch latch = new CountDownLatch(blockInfos.size());
			DataLocalityRunner.runWithLocality(blockInfos, new DataLocalityTask() {
				@Override
				public void run(BlockInfo blkInfo, String selectedHost) {
					try {
						URI uri = new URI("http", null, selectedHost, port, contextPath + "/DoSearch", "source="
								+ blkInfo.getFilePath() + "&start=" + blkInfo.getBlockLocation().getOffset() + "&end="
								+ (blkInfo.getBlockLocation().getLength() + blkInfo.getBlockLocation().getOffset() - 1)
								+ "&key=" + key + "&syncPath=" + syncPath, null);

						final HttpGet httpGet = new HttpGet(uri);
						httpclient.execute(httpGet, new FutureCallback<HttpResponse>() {
							public void completed(final HttpResponse resp) {
								latch.countDown();
								logger.info(httpGet.getRequestLine() + "->" + resp.getStatusLine());
								// TODO handle socket time out.
								try {
									String line = "";
									BufferedReader rd = new BufferedReader(new InputStreamReader(resp.getEntity()
											.getContent()));
									while ((line = rd.readLine()) != null) {
										wri.write(line);
									}
								} catch (IOException e) {
									logger.error("error:", e);
								}
							}

							public void failed(final Exception ex) {
								latch.countDown();
								// TODO handle socket timeout here, maybe retry or send the request to another host.
								logger.info(httpGet.getRequestLine() + "->" + ex);
							}

							public void cancelled() {
								latch.countDown();
								logger.info(httpGet.getRequestLine() + " cancelled");
							}
						});
					} catch (URISyntaxException e) {
						e.printStackTrace();
						throw new RuntimeException(e);
					}
				}
			});

			latch.await();
			logger.info("completed");
		} catch (InterruptedException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} finally {
			try {
				httpclient.close();
			} catch (IOException ee) {
			}
		}
	}

	protected void doPost(final HttpServletRequest request, final HttpServletResponse response)
			throws ServletException, IOException {
		final String key = request.getParameter("key");
		final String filePath = request.getParameter("file");
		logger.info("key=" + key);
		logger.info("filePath=" + filePath);

		//final String uniquePath = createZnode();

		doSearch(key, new String[] { filePath }, response.getWriter(), request.getContextPath(),
				request.getServerPort(), null);

		logger.info("Done");
	}


}
